package com.niodata.dt.fs.ttl;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * file lifecycle processor,non threadsafe.
 */
public class TtlProcessor {

  private static final Logger logger = Logger.getLogger(TtlProcessor.class);
  private volatile boolean stop = false;
  private TtlProcessStatistics statistics;
  private volatile boolean started = false;

  public TtlProcessor() {
    statistics = new TtlProcessStatistics();
  }

  /**
   * do file ttl process.
   *
   * @param fs filesystem
   * @param config ttl config
   * @throws IOException ex
   */
  public void doTtlDelete(FileSystem fs, TtlConfig config) throws IOException {
    if (!started || stop) {
      throw new RuntimeException("ttl processor not started or stopped");
    }
    List<Path> paths = new LinkedList();
    List<Path> removedPaths = new LinkedList<>();
    paths.add(new Path(config.getDir()));
    while (!paths.isEmpty()) {
      Path dir = ((LinkedList<Path>) paths).remove();
      statistics.increaseDirectoryCount(1);
      FileStatus[] statuses = fs.listStatus(dir);
      logger.info("check directory ttl start:" + dir);
      for (FileStatus status : statuses) {
        if (stop) {
          logger.info("ttl process interrupted...");
          return;
        }
        if (status.isFile()) {
          if (!status.getPath().getName().matches(config.getNamePattern())) {
            continue;
          }
          if (status.getModificationTime() < config.getMaxModificationTime()) {
            fs.delete(status.getPath(), false);
            statistics.increaseFileCount(1);
          }
        } else if (config.isRecursive() && fs.listStatus(status.getPath()).length > 0) {
          paths.add(status.getPath());
        }
      }
      logger.info("check directory ttl finished:" + dir);
      if (fs.exists(dir)) {
        //object store,dir maybe not exist if no files in it
        removedPaths.add(dir);
      }
    }
    //recheck empty paths;
    int removed = 0;
    for (Path dir : removedPaths) {
      if (fs.exists(dir) && fs.listStatus(dir).length == 0) {
        fs.delete(dir, false);
        removed++;
      }
    }
    logger.info("checked paths:" + removedPaths.size());
    logger.info("removed paths:" + removed);

    statistics.setEndTime(new Date());
  }

  /**
   * do file ttl process.
   *
   * @param fs filesystem
   * @param config ttl config
   * @throws IOException ex
   */
  public void doTtlMove(FileSystem fs, FileSystem targetFs, TtlConfig config)
        throws IOException {
    if (!started || stop) {
      throw new RuntimeException("ttl processor not started or stopped");
    }
    List<Path> paths = new LinkedList();
    List<Path> removedPaths = new LinkedList<>();
    paths.add(new Path(config.getDir()));
    while (!paths.isEmpty()) {
      Path dir = ((LinkedList<Path>) paths).remove();
      FileStatus[] statuses = fs.listStatus(dir);
      statistics.increaseDirectoryCount(1);
      logger.info("check directory ttl start:" + dir);
      for (FileStatus status : statuses) {
        if (stop) {
          logger.info("ttl process interrupted...");
          return;
        }
        if (status.isFile()) {
          if (!status.getPath().getName().matches(config.getNamePattern())) {
            continue;
          }
          if (status.getModificationTime() < config.getMaxModificationTime()) {
            String distDir = Path.getPathWithoutSchemeAndAuthority(status.getPath())
                  .toString().replace(new Path(config.getDir()).getParent().toString(), "");
            distDir = config.getTargetDir() + "/" + distDir;
            while (distDir.contains("//")) {
              distDir = distDir.replaceAll("//", "/");
            }
            Path out = new Path(distDir);
            if (!targetFs.exists(out.getParent())) {
              targetFs.mkdirs(out.getParent());
            }
            InputStream input = fs.open(status.getPath());
            OutputStream output = targetFs.create(out, true);
            IOUtils.copy(input, output);
            input.close();
            output.close();
            fs.delete(status.getPath(), false);
            statistics.increaseFileCount(1);
          }
        } else if (config.isRecursive() && fs.listStatus(status.getPath()).length > 0) {
          paths.add(status.getPath());
        }
      }
      logger.info("check directory ttl finished:" + dir);
      if (fs.exists(dir)) {
        //object store,dir maybe not exist if no files in it
        removedPaths.add(dir);
      }
    }
    //recheck empty paths;
    logger.info("removed paths:" + removedPaths.size());
    for (Path dir : removedPaths) {
      if (fs.exists(dir) && fs.listStatus(dir).length == 0) {
        fs.delete(dir, false);
      }
    }
  }

  public void start() {
    if (started || stop) {
      return;
    }
    this.started = true;
    this.statistics.setStartTime(new Date());
  }

  public void stop() {
    this.stop = true;
    this.statistics.setEndTime(new Date());
  }

  public void resetStatistics() {
    this.statistics = new TtlProcessStatistics();
  }

  public TtlProcessStatistics getStatistics() {
    return this.statistics;
  }
}
